package com.bz.service.socket;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.bz.config.IMConfig;
import com.bz.constans.Constants;
import com.bz.model.ChatMessage;
import com.bz.model.MessageListO;
import com.bz.service.impl.ChatService;
import com.bz.util.LoginUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

import com.bz.util.RedisUtil;
import org.thymeleaf.util.StringUtils;
import com.bz.util.UuidUtils;

/**
 * 群组聊天室连接（监听群内聊天消息更新）
 */
@Component
@ServerEndpoint("/group/chat/{groupId}/{userId}/{otherUser}/{token}")
public class GroupChatServer {

    //Redis工具类（交由IOC自动注入）
    private static RedisUtil redisUtil;

    //IM配置信息加载（交由IOC自动注入）
    private static IMConfig imConfig;

    //登录状态检查（交由IOC自动注入）
    private static LoginUtil loginUtil;

    //Redis订阅监听器设置（交由IOC自动注入）
    private static RedisMessageListenerContainer redisMessageListenerContainer;

    private static ChatService chatService;

    @Autowired
    public void setRepository(RedisUtil redisUtil) {
        GroupChatServer.redisUtil = redisUtil;
    }

    @Autowired
    public void setRepository(IMConfig imConfig) {
        GroupChatServer.imConfig = imConfig;
    }

    @Autowired
    public void setRepository(LoginUtil loginUtil) {
        GroupChatServer.loginUtil = loginUtil;
    }

    @Autowired
    public void setRepository(ChatService chatService) {
        GroupChatServer.chatService = chatService;
    }


    @Autowired
    public void setRepository(RedisMessageListenerContainer redisMessageListenerContainer) {
        GroupChatServer.redisMessageListenerContainer = redisMessageListenerContainer;
    }

    //Redis订阅推送消息服务（设为静态，仅在本类加载时调用一次）
    private static RedisListenServer redisListenServer = new RedisListenServer();

    //concurrent包的线程安全Set，用来存放每个客户端对应的WebSocketServer对象。
    private static ConcurrentHashMap<String, Session> sessionPools = new ConcurrentHashMap<>();

    //记录Redis监听器，避免不同用户连接同一群组时重复创建订阅监听。
    private static ConcurrentHashMap<String, Boolean> redisListenMap = new ConcurrentHashMap<>();

    //发送消息
    public static void sendMessage(Session session, String message) throws IOException {
        if (session != null) {
            synchronized (session) {
                session.getBasicRemote().sendText(message);
            }
        }
    }

    /**
     * 建立连接成功调用
     *
     * @param session 会话
     * @param userId  用户id
     * @param groupId 群组id
     */
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "token") String token,
                       @PathParam(value = "userId") String userId,
                       @PathParam(value = "otherUser") String otherUser,
                       @PathParam(value = "groupId") String groupId) throws IOException {
        try {
            // ---------------- 设置监听器，监听在线消息  ----------------
            //将session与userId传入Redis订阅监听器
            redisListenServer.setSession(userId, session);
            //设置监听器，监听推送该群组的消息，redisListenMap避免重复订阅。
            if (redisListenMap.get("mq:" + groupId) == null) {
                //如果当前这个"mq:"+groupId主题没有被订阅，则建立监听器。
                redisMessageListenerContainer.addMessageListener(redisListenServer, new PatternTopic("mq:" + groupId));
                redisListenMap.put("mq:" + groupId, true);
            }
            // ---------------- 设置监听器，监听在线消息  ----------------

            /*
             * === 返回群内最新的消息列表 ===
             */
            //单次获取的消息数量
            long listNum = imConfig.getListNum();
//            long count = chatService.count();
            List<ChatMessage> chatMessages = chatService.performSearch(groupId, 0, UuidUtils.safeLongToInt(listNum));
            //更新用户未读消息数量
            updateMessTotZJ(userId, groupId, otherUser);
            //发送消息（序列化发送）
            if (CollectionUtil.isEmpty(chatMessages)) {
                //todo message 可以从 otherUser 自定义取值
                ChatMessage chatMessage = new ChatMessage(groupId, otherUser, "自动给用户发送欢迎语（可以从 " + otherUser + " 自定义取值）");
                chatMessages.add(chatMessage);
            }
            sendMessage(session, JSON.toJSONString(chatMessages));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 关闭连接时调用
     *
     * @param userId 用户id
     */
    @OnClose
    public void onClose(@PathParam(value = "userId") String userId) {
        sessionPools.remove(userId);//删除用户
        redisUtil.removeMembers(Constants.SESSION_KEY_PREFIX, userId);
    }

    /**
     * 收到客户端信息
     *
     * @param message 消息主体内容
     * @param userId  用户id
     * @param groupId 群组id
     */
    @OnMessage
    public void onMessage(String message,
                          @PathParam(value = "token") String token,
                          @PathParam(value = "userId") String userId,
                          @PathParam(value = "otherUser") String otherUser,
                          @PathParam(value = "groupId") String groupId) {
        ChatMessage chatMessage = chatService.saveMessage(groupId, userId, message);
        updateMessTotZJ(userId, groupId, otherUser);
        updateMessTotOther(otherUser, groupId, userId);
        //在redis中发布消息
        redisUtil.pub("mq:" + groupId, JSON.toJSONString(chatMessage));
    }

    //错误时调用
    @OnError
    public void onError(Session session, Throwable throwable) throws IOException {
        throwable.printStackTrace();
        //关闭对话
        session.close();
    }

    /**
     * 获取消息总线
     */
    public List<MessageListO> getMessTot(String userId) {
        //连接建立后返回{listNum}条消息（从list右侧截取消息列表，获取最新的{listNum}条数据）
        List<Object> list = redisUtil.lGet(Constants.CHANNEL_ + userId, 0, -1);
        List<MessageListO> listOS = new ArrayList<>();
        Object o;
        MessageListO object;
        for (int i = 0; i < list.size(); i++) {
            o = list.get(i);
            object = (MessageListO) o;
            listOS.add(object);
        }
        return listOS;
    }

    /**
     * 更新消息总线计数器（当有消息发送的时候，异步的请求redis，让redis 更新用户b的消息列表）
     */
    @Async
    public void updateMessTotZJ(String userId, String roomId, String otherUser) {
        List<MessageListO> listOS = getMessTot(userId);
        MessageListO message;
        int a = 0;
        for (int i = 0; i < listOS.size(); i++) {
            message = listOS.get(i);
            if (StringUtils.equals(message.getRoomId(), roomId)) {
                message.setTotal(0);
                //更新
                redisUtil.lUpdateIndex(Constants.CHANNEL_ + userId, i, message);
                a = 1;
            }
        }
        if (a == 0) {
            // 插入
            message = new MessageListO();
            message.setRoomId(roomId);
            message.setTotal(0);
            message.setOtherUserId(otherUser);
            redisUtil.lSet(Constants.CHANNEL_ + userId, message, imConfig.getTimeout());
        }
    }


    @Async
    public void updateMessTotOther(String otherUser, String roomId, String userId) {
        int b = 0;
        List<MessageListO> listOS = getMessTot(otherUser);
        List<MessageListO> listOSNew = new ArrayList<>();
        MessageListO message = null;
        for (int i = 0; i < listOS.size(); i++) {
            message = listOS.get(i);
            if (StringUtils.equals(message.getRoomId(), roomId)) {
                message.setTotal(message.getTotal() + 1);
                //更新
                redisUtil.lUpdateIndex(Constants.CHANNEL_ + otherUser, i, message);
                b = 1;
            }
            listOSNew.add(message);
        }
        if (b == 0) {
            message = new MessageListO();
            message.setRoomId(roomId);
            message.setTotal(1);
            message.setOtherUserId(userId);
            redisUtil.lSet(Constants.CHANNEL_ + otherUser, message, imConfig.getTimeout());
        }
    }

}

